package com.atguigu.flink.exec2;

import com.atguigu.flink.pojo.AdsClickLog;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * Created by Smexy on 2023/2/2
 *
 * 统计当天用户对广告的点击量，并统计拉黑的用户
 *
 *      ①对用户在当天的行为进行约束
 *      ②点击某个广告超过100次，拉黑，当天的点击不统计
 *      ③第二天，一切重置
 *      
 *      
 *      变量:
 *              a) 存储是否已经拉黑的标记
 *              b) 存储当前累积的点击次数
 *              c) 当前的日期，判断日期是否变化
 *      
 */
public class Demo4_AdsBlack
{
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        //读数据
        env.readTextFile("data/AdClickLog.csv").setParallelism(1)
           .map(new MapFunction<String, AdsClickLog>()
           {
               @Override
               public AdsClickLog map(String value) throws Exception {
                   String[] words = value.split(",");
                   return new AdsClickLog(
                       Long.parseLong(words[0]),
                       Long.parseLong(words[1]),
                       words[2],
                       words[3],
                       Long.parseLong(words[4]) * 1000
                   );
               }
           }).setParallelism(1)
           .keyBy(adsClickLog -> adsClickLog.getUserId() + "-" + adsClickLog.getAdsId())
           .process(new KeyedProcessFunction<String, AdsClickLog, String>()
           {

               private ValueState<Integer> click;
               private ValueState<String> lastDate;
               private ValueState<Boolean> ifBlack;

               @Override
               public void open(Configuration parameters) throws Exception {
                   ifBlack = getRuntimeContext().getState(new ValueStateDescriptor<>("ifBlack", Boolean.class));
                   click = getRuntimeContext().getState(new ValueStateDescriptor<>("click", Integer.class));
                   lastDate = getRuntimeContext().getState(new ValueStateDescriptor<>("lastDate", String.class));
               }

               //不用，只读
               private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");

               @Override
               public void processElement(AdsClickLog value, Context ctx, Collector<String> out) throws Exception {

                   String currentDate = this.format.format(new Date(value.getTimestamp()));
                   //要求：数据必须按照时间顺序到达。 判断当前数据的日期是否已经跨天
                   if (lastDate.value() == null || !currentDate.equals(lastDate.value())){
                       //第一条数据到达 或 日期变化，都初始化统计指标(重置)
                       ifBlack.update(false);
                       click.update(0);
                   }

                   //更新日期
                   lastDate.update(currentDate);

                   //累计点击次数
                   click.update(click.value() + 1);

                   //取出当前累计的次数
                   Integer clickCount = click.value();

                   //判断是否符合拉黑的条件
                   if ( !ifBlack.value() &&  clickCount > 100){

                       //符合拉黑条件
                       out.collect(currentDate+"  " +ctx.getCurrentKey() +" 超过100次，拉黑");
                       ifBlack.update(true);

                   }else{

                       if (clickCount <= 100){

                           out.collect(currentDate+"  " +ctx.getCurrentKey() +" "+clickCount+"次");
                       }

                   }



               }
           })
           .print().setParallelism(1);

        env.execute();
    }
}
